import kafka from "node-rdkafka"
import { eventHandler, EventCode } from "../event/eventHandler.js"
import {parseConf,conf} from "../confParse"

async function kafkaCreateTopic(conf: conf) {
    const client = kafka.AdminClient.create({
        'client.id': 'kafka-admin',
        'metadata.broker.list': conf.brokerList
    });
    try {
       await createTopic(client,conf.reqTopic,conf.instanceNum)
       if(conf.logType==="kafka"){
        await createTopic(client,"nodeLogTopic",1)
        await createTopic(client,"nodeErrLogTopic",1)
       }
       
        client.disconnect()
        // switch(conf)

    } catch (err) {
        eventHandler.handle(EventCode.CONNECT_KAFKA_ERROR,
            "for admin" + conf.brokerList,
            err as Error)
    }
}

async function createTopic(client: kafka.IAdminClient,
    topicName: string,
    partitionsNum: number): Promise<void> {
    return new Promise(async (resolve, reject) => {
        client.createTopic({
            'topic': topicName,
            "replication_factor": 1,
            "num_partitions": partitionsNum,
        }, async (err) => {
            if (err === undefined ) {
                resolve
            }else if( err.code === 36){
                await new Promise<void>((res,rej)=>{
                    client.createPartitions(topicName,partitionsNum,(err)=>{
                        if(!err){
                            res()
                        }
                        rej(err)
                    })
                }).catch((err)=>{
                    reject(err)
                })
            }
            reject(err)
        })
    })
}


